package com.johnguo.beam_test.write;

import java.util.Arrays;
import java.util.List;

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.io.kafka.KafkaIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
import org.apache.kafka.common.serialization.StringSerializer;

import com.johnguo.beam_test.write.producer.ProducerFactoryFn;

public class KafkaDemoBeam {
	static final List<String> LINES1 = Arrays.asList("johnguo1", "johnguo2", "johnguo3", "johnguo4", "johnguo5");

	public static void main(String... args) {
		PipelineOptions options = PipelineOptionsFactory.fromArgs(args).withValidation().create();
		// System.out.println(options);
		Pipeline p = Pipeline.create(options);
		KafkaIO.Write<String, String> writer = KafkaIO.<String, String>write().withBootstrapServers("flink-master:9092")
				.withTopic("test").withProducerFactoryFn(new ProducerFactoryFn())
				.withKeySerializer(StringSerializer.class).withValueSerializer(StringSerializer.class);
		// .withMaxNumRecords(1);
		p.apply(Create.of(LINES1)).setCoder(StringUtf8Coder.of())
				.apply(ParDo.of(new DoFn<String, KV<String, String>>() {
					private static final long serialVersionUID = 1L;

					@ProcessElement
					public void processElement(ProcessContext c) {
						c.output(KV.of(c.element(), c.element()));
					}
				})).apply(writer);
		p.run().waitUntilFinish();
	}
}